/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: cbus_client_endpoint.c
 *
 * Purpose: client endpoint implementation for cbus.
 *
 * Developer:
 *   wen.gu , 2022-02-15
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/
#include "cbus/bus/cbus_endpoint_ops.h"

#include <string.h>

//#include "cbus/core/cbus_mutex.h"
#include "cbus/core/cbus_map.h"
#include "cbus/core/cbus_worker.h"
#include "cbus/core/cbus_thread.h"
#include "cbus/bus/cbus_dialer.h"
#include "cbus/bus/cbus_pipe.h"
#include "cbus/core/cbus_string.h"
#include "cbus/core/cbus_log.h"
/******************************************************************************
 **    MACROS
 ******************************************************************************/

#define DIALER_CONNECT_WAIT_TIME (1000)
#define DIALER_CONNECT_TRY_COUNT (3)

#define PIPE_RECV_BUF_SIZE (1024 * 16)
#define PIPE_RECV_WAIT_TIME (1000)

/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/
typedef struct _cbus_client_endpoint
{
    GSTR url;
    GPTR on_connect_opaque; /** a user opaque pointor for on connect changed */
    GPTR on_recv_opaque;
    cbus_endpoint_on_connect_changed_func on_connect;
    cbus_endpoint_on_receive_func on_recv;
    cbus_dialer_t dialer;  
    cbus_pipe_t com_pipe; /** communication pipe with service */
    cbus_worker_t receive_worker;
}cbus_client_endpoint_t;

/******************************************************************************
 **    inner FUNCTION DEFINITIONS
 ******************************************************************************/

static void close_client_pipe(cbus_client_endpoint_t* client)
{
    if (client->com_pipe)
    {
        cbus_pipe_destroy(client->com_pipe);        
        client->com_pipe = NULL;
    }    
}

static void connect_status_notify(cbus_client_endpoint_t* client, GBOL connected)
{
    if (client->on_connect)
    {
        client->on_connect(client->on_connect_opaque, NULL, connected);
    }   

    if (connected == GFALSE)
    {
        close_client_pipe(client);
    }
}


static void cbus_client_endpoint_free(cbus_client_endpoint_t* client)
{
    cbus_worker_join(client->receive_worker);
    cbus_pipe_destroy(client->com_pipe);       
    cbus_dialer_destroy(client->dialer); //todo something

    if (client->url)
    {
        free(client->url);
    }

    free(client);
}


static void client_ep_on_receive(cbus_worker_t thd_worker, void* ptr)
{
    GU08 recv_buf[PIPE_RECV_BUF_SIZE];
    cbus_client_endpoint_t* client = (cbus_client_endpoint_t*)ptr;

    while (cbus_worker_running(thd_worker))
    {
        if (!client->com_pipe)
        {
            cbus_thread_sleep(10);
            continue;
        }

        GS32 ret = cbus_pipe_read(client->com_pipe, recv_buf, PIPE_RECV_BUF_SIZE, PIPE_RECV_WAIT_TIME);

        if (ret > 0)
        {
            if (client->on_recv)
            {
                client->on_recv(client->on_recv_opaque, client->com_pipe, recv_buf, ret);
            }
        }
        else if ((G_ErrInterrupted == ret) || (G_ErrTimeout == ret))
        {
            //do nothing ,just try again
        }
        else 
        {
            connect_status_notify(client, GFALSE);
            break; //todo refine me?? 
        }        
    }
}

/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/

static GErrc cbus_client_endpoint_initialize(GPHD* handle, const GSTR url)
{
   if (!handle || !url)
    {
        return G_ErrBadParameter;
    }
    cbus_transport_scheme_t scheme;
    GErrc ret = cbus_transport_get_scheme_name_by_url(url, scheme);

    if (G_OK != ret)
    {
        LOGE("get transport scheme by url failed(0x%08x)\n", scheme);
        return ret;
    }

    cbus_client_endpoint_t* client = (cbus_client_endpoint_t*)malloc(sizeof(cbus_client_endpoint_t));

    if (!client)
    {
        LOGE("alloc client endpoint instance failed\n");
        return G_ErrInsufficientResources;
    }

    memset(client, 0, sizeof(cbus_client_endpoint_t));

    ret = cbus_dialer_create(&client->dialer, scheme);

    if (G_OK != ret)
    {
        LOGE("create dialer instance by scheme(%s) failed(0x%08x)\n", scheme, ret);
        cbus_client_endpoint_free(client);
        return ret;
    }

    ret = cbus_worker_create(&client->receive_worker, "cli_ep_recv", client_ep_on_receive, client);
    
    if (G_OK != ret)
    {
        LOGE("create receive worker failed(0x%08x)\n", ret);
        cbus_client_endpoint_free(client);
        return ret;
    }

    client->url = cbus_str_dup(url);
    *handle = (GPHD)client;

    return G_OK;
}

static void cbus_client_endpoint_finish(GPHD handle)
{
    if (handle)
    {
        cbus_client_endpoint_free((cbus_client_endpoint_t*)handle);
    }
}

static GErrc cbus_client_endpoint_start(GPHD handle)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    cbus_client_endpoint_t* client = (cbus_client_endpoint_t*)handle;

    if (!client->url || client->com_pipe)
    {
        return G_ErrInvalidStatus;
    }

    GErrc ret = G_ErrTimeout;
    GU32 try_idx = 0;

    while (try_idx++ < DIALER_CONNECT_TRY_COUNT)
    {
        ret = cbus_dialer_connect(client->dialer, client->url, DIALER_CONNECT_WAIT_TIME, &client->com_pipe);

        if (G_OK == ret)
        {
            connect_status_notify(client, GTRUE);

            return cbus_worker_start(client->receive_worker);
        }
    }
    
    return ret;
}

static GErrc cbus_client_endpoint_stop(GPHD handle)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }
    
    cbus_client_endpoint_t* client = (cbus_client_endpoint_t*)handle;

    if (!client->com_pipe)
    {
        return G_ErrInvalidStatus;
    }

    cbus_pipe_destroy(client->com_pipe);
    cbus_worker_pause(client->receive_worker);

    client->com_pipe = NULL;
    return G_OK;
}

static GErrc cbus_client_endpoint_set_on_connect(GPHD handle, cbus_endpoint_on_connect_changed_func on_connect, GPTR opaque)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    cbus_client_endpoint_t* client = (cbus_client_endpoint_t*)handle;

    client->on_connect = on_connect;
    client->on_connect_opaque = opaque;
    return G_OK;
}

static GErrc cbus_client_endpoint_set_on_receive(GPHD handle, cbus_endpoint_id endpoint_id, cbus_endpoint_on_receive_func on_receive, GPTR opaque)
{
    if (!handle || !on_receive)
    {
        return G_ErrBadParameter;
    }

    (void)endpoint_id; //todo refine me??
    
    cbus_client_endpoint_t* client = (cbus_client_endpoint_t*)handle;
    client->on_recv        = on_receive;
    client->on_recv_opaque = opaque;

    return G_OK;
}

static GErrc cbus_client_endpoint_send(GPHD handle, cbus_endpoint_id endpoint_id, const GBUF data, GU32 size)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    cbus_client_endpoint_t* client = (cbus_client_endpoint_t*)handle;

    if (!client->com_pipe)
    {
        return G_ErrInvalidStatus;
    }

    (void)endpoint_id; /** the client side needn't care about endpoint id */

    GErrc ret = cbus_pipe_write(client->com_pipe, data, size);
    //LOGD("[%s.%d]===>: pipe write return: 0x%08x, size: %d\n", __FUNCTION__, __LINE__, ret, size);
    if (G_ErrDisconnected == ret)
    {
        connect_status_notify(client, GFALSE);        
    }

    return ret;
}






//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

static const cbus_endpoint_ops g_client_endpoint_ops = 
{
    CBUS_EP_ROLE_CLIENT,
    CBUS_EP_VERSION,
    "default",
    cbus_client_endpoint_initialize,
    cbus_client_endpoint_finish,
    cbus_client_endpoint_start,
    cbus_client_endpoint_stop,
    cbus_client_endpoint_set_on_connect,
    cbus_client_endpoint_set_on_receive,
    cbus_client_endpoint_send,
};

GErrc cbus_client_endpoint_default_ctor(void)
{
    return cbus_endpoint_ops_register(&g_client_endpoint_ops);
    //return G_OK;
}
